import datetime
import json
import time

from dateutil.tz import tzutc
from kubernetes import client, watch, config
from kubernetes.client.rest import ApiException

from airflow import AirflowException



kubeconfig_path = "/Users/lianbin/.kube/config"
# kubeconfig_path = "/root/.kube/config"
config.load_kube_config(config_file=kubeconfig_path)
# 创建 Kubernetes API 客户端
v1 = client.AppsV1Api()

w = watch.Watch()
hostname = "my-first-application-cluster"
rv = '0'
stop = False

while(True):
    try:
        rv = '0'
        retry_timestamp = int(time.time())
        j_retry = 0
        watch_event = w.stream(v1.list_namespaced_deployment, namespace="dm-poc",
                               label_selector="app=task-33-flink-cluster",
                               resource_version=rv )
        for event in watch_event:
            ready_replicas = event['object'].status.ready_replicas
            event_type = event['type']
            print("flink app running status:{}, ready_replicas:{}".format(event_type, ready_replicas))
            if event_type == 'ERROR':
                rv = '0'
                print("update resource version")
                continue
            new_version = event['object'].metadata.resource_version
            if not rv or int(rv) < int(new_version):
                rv = new_version
            if event_type == 'DELETED':
                w.stop()
                raise AirflowException(
                    "flink app error app: {}.replicas:{}".format(1, ready_replicas)
                )
            else:
                retry_interval = int(time.time()) - retry_timestamp
                if j_retry > 6 and retry_interval < 1800:
                    # 重试超过5次,事件间隔再半小时内杀死进程
                    # v1.delete_namespaced_deployment(namespace=self._namespace, name=cluster_id)
                    w.stop()
                    raise AirflowException(
                        "deployment is not ready: {}.".format(
                            1
                        )
                    )
                if event_type == 'MODIFIED' and ready_replicas == None:
                    current_timestamp = int(time.time())
                    if (current_timestamp - retry_timestamp) > 1800:
                        retry_timestamp = current_timestamp
                        j_retry = 0
                    j_retry += 1
        if stop:
            break
    except ApiException as e:
        print("ex")



w.stop()
print("end")

